<html>
<head>
  <title>Evernote Export</title>
  <basefont face="微软雅黑" size="2" />
  <meta http-equiv="Content-Type" content="text/html;charset=utf-8" />
  <meta name="exporter-version" content="YXBJ Windows/601748 (zh-CN, DDL); Windows/10.0.0 (Win64); EDAMVersion=V2;"/>
  <style>
    body, td {
      font-family: 微软雅黑;
      font-size: 14pt;
    }
  </style>
</head>
<body>
<a name="819"/>

<div>
<span><div><img src="Evernote_files/Image.png" type="image/png" data-filename="Image.png"/></div><div>*号:匹配1个单词, # 匹配1到多个单词</div><div><img src="Evernote_files/Image [1].png" type="image/png" data-filename="Image.png"/></div><div>实质 是对routing模式的增强,routing 每一个消息从交换机中发出,只能对应一个队列,topic 则通过routingKey 指定消息 对应 通配符表示的某一类队列</div><div>demo 结构</div><div><img src="Evernote_files/Image [2].png" type="image/png" data-filename="Image.png"/></div><div><br/></div><div>生产者代码</div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div>/**</div><div>* 发送消息</div><div>*/</div><div>public class TopicsProducer {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div><br/></div><div><br/></div><div>        //1.创建连接工厂</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2. 设置参数</div><div>        factory.setHost(&quot;localhost&quot;);//ip  默认值 localhost</div><div>        factory.setPort(5672); //端口  默认值 5672</div><div>        factory.setVirtualHost(&quot;/itcast&quot;);//虚拟机 默认值/</div><div>        factory.setUsername(&quot;heima&quot;);//用户名 默认 guest</div><div>        factory.setPassword(&quot;heima&quot;);//密码 默认值 guest</div><div>        //3. 创建连接 Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div>       /*</div><div><br/></div><div><br/></div><div>       exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map&lt;String, Object&gt; arguments)</div><div>       参数：</div><div>        1. exchange:交换机名称</div><div>        2. type:交换机类型</div><div>            DIRECT(&quot;direct&quot;),：定向</div><div>            FANOUT(&quot;fanout&quot;),：扇形（广播），发送消息到每一个与之绑定队列。</div><div>            TOPIC(&quot;topic&quot;),通配符的方式</div><div>            HEADERS(&quot;headers&quot;);参数匹配</div><div><br/></div><div><br/></div><div>        3. durable:是否持久化</div><div>        4. autoDelete:自动删除</div><div>        5. internal：内部使用。 一般false</div><div>        6. arguments：参数</div><div>        */</div><div><br/></div><div><br/></div><div>       String exchangeName = &quot;test_topic&quot;;</div><div>        //5. 创建交换机</div><div>        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);</div><div>        //6. 创建队列</div><div>        String queue1Name = &quot;test_topic_queue1&quot;;</div><div>        String queue2Name = &quot;test_topic_queue2&quot;;</div><div>        channel.queueDeclare(queue1Name,true,false,false,null);</div><div>        channel.queueDeclare(queue2Name,true,false,false,null);</div><div>        //7. 绑定队列和交换机</div><div>        /*</div><div>        queueBind(String queue, String exchange, String routingKey)</div><div>        参数：</div><div>            1. queue：队列名称</div><div>            2. exchange：交换机名称</div><div>            3. routingKey：路由键，绑定规则</div><div>                如果交换机的类型为fanout ，routingKey设置为&quot;&quot;</div><div>         */</div><div><br/></div><div><br/></div><div>        // routing key  系统的名称.日志的级别。</div><div>        //=需求： 所有error级别的日志存入数据库，所有order系统的日志存入数据库</div><div>        channel.queueBind(queue1Name,exchangeName,&quot;#.error&quot;);</div><div>        channel.queueBind(queue1Name,exchangeName,&quot;order.*&quot;);</div><div>        channel.queueBind(queue2Name,exchangeName,&quot;*.*&quot;);</div><div><br/></div><div><br/></div><div>        String body = &quot;日志信息：张三调用了findAll方法...日志级别：info...&quot;;</div><div>        //8. 发送消息</div><div>        channel.basicPublish(exchangeName,&quot;goods.error&quot;,null,body.getBytes());</div><div><br/></div><div><br/></div><div>        //9. 释放资源</div><div>        channel.close();</div><div>        connection.close();</div><div><br/></div><div><br/></div><div>    }</div></div><div>消费者1代码</div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div>public class Consumer_Topic1 {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div><br/></div><div><br/></div><div>        //1.创建连接工厂,获取连接</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2.设置参数 ,不设置则使用默认值localhost</div><div>        factory.setHost(&quot;192.168.170.131&quot;);</div><div>        //端口 默认 5672</div><div>        factory.setPort(5672);</div><div>        //虚拟机,默认值 &quot;/&quot;</div><div>        factory.setVirtualHost(&quot;/myRabbit&quot;);</div><div>        //用户名密码</div><div>        factory.setUsername(&quot;gzx&quot;);</div><div>        factory.setPassword(&quot;gzx&quot;);</div><div>        //3. 创建连接Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div><br/></div><div><br/></div><div><br/></div><div><br/></div><div>        String queue1Name = &quot;test_topic_queue1&quot;;</div><div>        String queue2Name = &quot;test_topic_queue2&quot;;</div><div><br/></div><div><br/></div><div><br/></div><div><br/></div><div>        /*</div><div>        basicConsume(String queue, boolean autoAck, Consumer callback)</div><div>        参数：</div><div>            1. queue：队列名称</div><div>            2. autoAck：是否自动确认</div><div>            3. callback：回调对象</div><div><br/></div><div><br/></div><div>         */</div><div>        // 接收消息</div><div>        Consumer consumer = new DefaultConsumer(channel){</div><div>            /*</div><div>                回调方法，当收到消息后，会自动执行该方法</div><div><br/></div><div><br/></div><div>                1. consumerTag：标识</div><div>                2. envelope：获取一些信息，交换机，路由key...</div><div>                3. properties:配置信息</div><div>                4. body：数据</div><div><br/></div><div><br/></div><div>             */</div><div>            @Override</div><div>            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {</div><div>              /*  System.out.println(&quot;consumerTag：&quot;+consumerTag);</div><div>                System.out.println(&quot;Exchange：&quot;+envelope.getExchange());</div><div>                System.out.println(&quot;RoutingKey：&quot;+envelope.getRoutingKey());</div><div>                System.out.println(&quot;properties：&quot;+properties);*/</div><div>                System.out.println(&quot;body：&quot;+new String(body));</div><div>                System.out.println(&quot;将日志信息存入数据库.......&quot;);</div><div>            }</div><div>        };</div><div>        channel.basicConsume(queue1Name,true,consumer);</div><div><br/></div><div><br/></div><div><br/></div><div><br/></div><div>        //关闭资源？不要</div><div><br/></div><div><br/></div><div>    }</div></div><div>消费者2代码</div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div>public class Consumer_Topic2 {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div><br/></div><div><br/></div><div>        //1.创建连接工厂,获取连接</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2.设置参数 ,不设置则使用默认值localhost</div><div>        factory.setHost(&quot;192.168.170.131&quot;);</div><div>        //端口 默认 5672</div><div>        factory.setPort(5672);</div><div>        //虚拟机,默认值 &quot;/&quot;</div><div>        factory.setVirtualHost(&quot;/myRabbit&quot;);</div><div>        //用户名密码</div><div>        factory.setUsername(&quot;gzx&quot;);</div><div>        factory.setPassword(&quot;gzx&quot;);</div><div>        //3. 创建连接Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div><br/></div><div><br/></div><div><br/></div><div><br/></div><div>        String queue1Name = &quot;test_topic_queue1&quot;;</div><div>        String queue2Name = &quot;test_topic_queue2&quot;;</div><div><br/></div><div><br/></div><div><br/></div><div><br/></div><div>        /*</div><div>        basicConsume(String queue, boolean autoAck, Consumer callback)</div><div>        参数：</div><div>            1. queue：队列名称</div><div>            2. autoAck：是否自动确认</div><div>            3. callback：回调对象</div><div><br/></div><div><br/></div><div>         */</div><div>        // 接收消息</div><div>        Consumer consumer = new DefaultConsumer(channel){</div><div>            /*</div><div>                回调方法，当收到消息后，会自动执行该方法</div><div><br/></div><div><br/></div><div>                1. consumerTag：标识</div><div>                2. envelope：获取一些信息，交换机，路由key...</div><div>                3. properties:配置信息</div><div>                4. body：数据</div><div><br/></div><div><br/></div><div>             */</div><div>            @Override</div><div>            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {</div><div>              /*  System.out.println(&quot;consumerTag：&quot;+consumerTag);</div><div>                System.out.println(&quot;Exchange：&quot;+envelope.getExchange());</div><div>                System.out.println(&quot;RoutingKey：&quot;+envelope.getRoutingKey());</div><div>                System.out.println(&quot;properties：&quot;+properties);*/</div><div>                System.out.println(&quot;body：&quot;+new String(body));</div><div>                System.out.println(&quot;将日志信息打印控制台.......&quot;);</div><div>            }</div><div>        };</div><div>        channel.basicConsume(queue2Name,true,consumer);</div><div><br/></div><div><br/></div><div><br/></div><div><br/></div><div>        //关闭资源？不要</div><div><br/></div><div><br/></div><div>    }</div><div>}</div></div><div>效果图</div><div><img src="Evernote_files/Image [3].png" type="image/png" data-filename="Image.png"/></div><div><br/></div></span>
</div>
<hr>
<a name="799"/>

<div>
<span><div><img src="Evernote_files/Image [4].png" type="image/png" data-filename="Image.png"/></div><div>需求:根据信息携带的不同标识发送到不同的队列 进行对应的逻辑处理</div><div>demo 结构</div><div><img src="Evernote_files/Image [5].png" type="image/png" data-filename="Image.png"/></div><div><br/></div><div>生产者代码</div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div>/**</div><div>* 发送消息</div><div>* @author GaoZeXi</div><div>*/</div><div>public class RoutingProducer {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div>        //1.创建连接工厂,获取连接</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2.设置参数 ,不设置则使用默认值localhost</div><div>        factory.setHost(&quot;192.168.170.131&quot;);</div><div>        //端口 默认 5672</div><div>        factory.setPort(5672);</div><div>        //虚拟机,默认值 &quot;/&quot;</div><div>        factory.setVirtualHost(&quot;/myRabbit&quot;);</div><div>        //用户名密码</div><div>        factory.setUsername(&quot;gzx&quot;);</div><div>        factory.setPassword(&quot;gzx&quot;);</div><div>        //3. 创建连接Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div>        //5. 创建交换机</div><div>        /**</div><div>         * exchangeDeclare(String exchange, 交换机名称</div><div>         *         BuiltinExchangeType type, String类型,或者枚举类型 (direct 定向发送消息,fanout 广播发送,</div><div>         *         topic 通配符方式,headers 参数匹配)</div><div>         *         boolean durable, 是否持久化</div><div>         *         boolean autoDelete, 当没有消费者连接后是否自动删除</div><div>         *         boolean internal,内部的, MQ使用,比如erlang语言内部插件调用</div><div>         *         Map&lt;String, Object&gt; arguments) 参数</div><div>         */</div><div>        String exchangeName =&quot;HelloWorld_routing_direct&quot;;</div><div>        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,null);</div><div>        //6. 创建队列</div><div>        /**</div><div>         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,</div><div>         *                                  Map&lt;String, Object&gt; arguments);</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. durable:是否持久化</div><div>         *      3. exclusive:</div><div>         *              是否独占,只能有一个消费者监听这个队列</div><div>         *              当Connection关闭时,是否删除队列</div><div>         *      4. autoDelete:是否自动删除,当没有Consumer来消费时,自动删除掉,</div><div>         *      5. arguments:参数.</div><div>         */</div><div>        String queue1Name = &quot;routing_direct_queue1&quot;;</div><div>        String queue2Name = &quot;routing_direct_queue2&quot;;</div><div>        //如果没有一个名字叫hello_world的队列,就创建它,有则不创建</div><div>        channel.queueDeclare(queue1Name,true,false,false,null);</div><div>        channel.queueDeclare(queue2Name,true,false,false,null);</div><div>        //7. 队列和交换机绑定</div><div>        /**</div><div>         * queueBind(String queue, 队列名称</div><div>         *          String exchange, 交换机名称</div><div>         *          String routingKey 路由键 ,即绑定规则 ,fanout 模式下 routingKey为&quot;&quot;)</div><div>         */</div><div>        channel.queueBind(queue1Name,exchangeName,&quot;error&quot;);</div><div>        channel.queueBind(queue2Name,exchangeName,&quot;info&quot;);</div><div>        channel.queueBind(queue2Name,exchangeName,&quot;error&quot;);</div><div>        channel.queueBind(queue2Name,exchangeName,&quot;warning&quot;);</div><div>        //6. 发送消息</div><div>        /**</div><div>         * basicPublish(String exchange, String routingKey, boolean mandatory,</div><div>         *                          boolean immediate, AMQP.BasicProperties props, byte[] body)</div><div>         * 参数:</div><div>         *      1. exchange:交换机名称,简单模式下,只能使用默认交换机</div><div>         *      2. routingKey:路由名称</div><div>         *      3. prop:配置信息</div><div>         *      4. body:发送的消息数据,</div><div>         */</div><div>        int msgNumber =10;</div><div>        for (int i = 1; i &lt;= msgNumber ; i++) {</div><div>            String body =&quot;routing _msg&quot;+i;</div><div>            if(i%2==0){</div><div>                channel.basicPublish(exchangeName,&quot;warning&quot;,null,body.getBytes());</div><div>            }</div><div>            if(i%3==0){</div><div>                channel.basicPublish(exchangeName,&quot;error&quot;,null,body.getBytes());</div><div>            }</div><div>            channel.basicPublish(exchangeName,&quot;info&quot;,null,body.getBytes());</div><div>        }</div><div>        channel.close();</div><div>        connection.close();</div><div>    }</div><div>}</div></div><div>消费者1代码</div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div>/**</div><div>* 接受消息</div><div>* @author GaoZeXi</div><div>*/</div><div>public class RoutingConsumer1 {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div>        //1.创建连接工厂,获取连接</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2.设置参数 ,不设置则使用默认值localhost</div><div>        factory.setHost(&quot;192.168.170.131&quot;);</div><div>        //端口 默认 5672</div><div>        factory.setPort(5672);</div><div>        //虚拟机,默认值 &quot;/&quot;</div><div>        factory.setVirtualHost(&quot;/myRabbit&quot;);</div><div>        //用户名密码</div><div>        factory.setUsername(&quot;gzx&quot;);</div><div>        factory.setPassword(&quot;gzx&quot;);</div><div>        //3. 创建连接Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div>        //5. 创建队列</div><div>        /**</div><div>         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,</div><div>         *                                  Map&lt;String, Object&gt; arguments);</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. durable:是否持久化</div><div>         *      3. exclusive:</div><div>         *              是否独占,只能有一个消费者监听这个队列</div><div>         *              当Connection关闭时,是否删除队列</div><div>         *      4. autoDelete:是否自动删除,当没有Consumer来消费时,自动删除掉,</div><div>         *      5. arguments:参数.</div><div>         */</div><div>        //如果没有一个名字叫hello_world的队列,就创建它,有则不创建</div><div>        //channel.queueDeclare(&quot;work_queues&quot;,true,false,false,null);</div><div>        //6. 接受消息</div><div>        /**</div><div>         * basicConsume(String queue, boolean autoAck, Consumer callback)</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. autoAck:是否自动确认</div><div>         *      3. callback:回调对象</div><div>         */</div><div>        String queue1Name = &quot;routing_direct_queue1&quot;;</div><div><br/></div><div><br/></div><div>        Consumer consumer = new DefaultConsumer(channel){</div><div>            /**</div><div>             * 回调方法,当收到消息后,会自动执行该方法</div><div>             * @param consumerTag 消息标识</div><div>             * @param envelope 获取一些信息,比如 交换机,routingKey...</div><div>             * @param properties 配置信息</div><div>             * @param body 消息体</div><div>             * @throws IOException</div><div>             */</div><div>            @Override</div><div>            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {</div><div>//                System.out.println(&quot;consumerTag:&quot;+consumerTag);</div><div>//                System.out.println(&quot;getExchange:&quot;+envelope.getExchange());</div><div>//                System.out.println(&quot;日志级别:&quot;+envelope.getRoutingKey());</div><div>//                System.out.println(&quot;getDeliveryTag:&quot;+envelope.getDeliveryTag());</div><div>//                System.out.println(&quot;properties:&quot;+properties);</div><div>                System.out.println(&quot;日志级别:&quot;+envelope.getRoutingKey()+&quot; 保存在数据库:&quot;+new String(body));</div><div>            }</div><div>        };</div><div>        //监听队列.消费数据</div><div>        channel.basicConsume(queue1Name,true,consumer);</div><div>//        消费者需要一直监听消息队列,不需要关闭!</div><div>//        channel.close();</div><div>//        connection.close();</div><div>    }</div><div>}</div></div><div>消费者2代码</div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div>/**</div><div>* 接受消息</div><div>* @author GaoZeXi</div><div>*/</div><div>public class RoutingConsumer2 {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div>        //1.创建连接工厂,获取连接</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2.设置参数 ,不设置则使用默认值localhost</div><div>        factory.setHost(&quot;192.168.170.131&quot;);</div><div>        //端口 默认 5672</div><div>        factory.setPort(5672);</div><div>        //虚拟机,默认值 &quot;/&quot;</div><div>        factory.setVirtualHost(&quot;/myRabbit&quot;);</div><div>        //用户名密码</div><div>        factory.setUsername(&quot;gzx&quot;);</div><div>        factory.setPassword(&quot;gzx&quot;);</div><div>        //3. 创建连接Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div>        //5. 创建队列</div><div>        /**</div><div>         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,</div><div>         *                                  Map&lt;String, Object&gt; arguments);</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. durable:是否持久化</div><div>         *      3. exclusive:</div><div>         *              是否独占,只能有一个消费者监听这个队列</div><div>         *              当Connection关闭时,是否删除队列</div><div>         *      4. autoDelete:是否自动删除,当没有Consumer来消费时,自动删除掉,</div><div>         *      5. arguments:参数.</div><div>         */</div><div>        //如果没有一个名字叫hello_world的队列,就创建它,有则不创建</div><div>        //channel.queueDeclare(&quot;work_queues&quot;,true,false,false,null);</div><div>        //6. 接受消息</div><div>        /**</div><div>         * basicConsume(String queue, boolean autoAck, Consumer callback)</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. autoAck:是否自动确认</div><div>         *      3. callback:回调对象</div><div>         */</div><div>        String queue2Name = &quot;routing_direct_queue2&quot;;</div><div><br/></div><div><br/></div><div>        Consumer consumer = new DefaultConsumer(channel){</div><div>            /**</div><div>             * 回调方法,当收到消息后,会自动执行该方法</div><div>             * @param consumerTag 消息标识</div><div>             * @param envelope 获取一些信息,比如 交换机,routingKey...</div><div>             * @param properties 配置信息</div><div>             * @param body 消息体</div><div>             * @throws IOException</div><div>             */</div><div>            @Override</div><div>            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {</div><div>//                System.out.println(&quot;consumerTag:&quot;+consumerTag);</div><div>//                System.out.println(&quot;getExchange:&quot;+envelope.getExchange());</div><div>//                System.out.println(&quot;getRoutingKey:&quot;+envelope.getRoutingKey());</div><div>//                System.out.println(&quot;getDeliveryTag:&quot;+envelope.getDeliveryTag());</div><div>//                System.out.println(&quot;properties:&quot;+properties);</div><div>                System.out.println(&quot;日志级别:&quot;+envelope.getRoutingKey()+&quot; 打印到控制台 :&quot;+new String(body));</div><div>            }</div><div>        };</div><div>        //监听队列.消费数据</div><div>        channel.basicConsume(queue2Name,true,consumer);</div><div>//        消费者需要一直监听消息队列,不需要关闭!</div><div>//        channel.close();</div><div>//        connection.close();</div><div>    }</div><div>}</div></div><div>效果图</div><div><img src="Evernote_files/Image [6].png" type="image/png" data-filename="Image.png"/><img src="Evernote_files/Image [7].png" type="image/png" data-filename="Image.png"/></div><div><br/></div></span>
</div>
<hr>
<a name="787"/>

<div>
<span><div><img src="Evernote_files/Image [8].png" type="image/png" data-filename="Image.png"/></div><div>需求:生产者生产一条消息,两个消费者都要接收,各自的处理逻辑不一样</div><div>demo 结构图</div><div><img src="Evernote_files/Image [9].png" type="image/png" data-filename="Image.png"/></div><div>生产者 代码</div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div><br/></div><div>/**</div><div>* 发送消息</div><div>* @author GaoZeXi</div><div>*/</div><div>public class PubSubProducer {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div>        //1.创建连接工厂,获取连接</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2.设置参数 ,不设置则使用默认值localhost</div><div>        factory.setHost(&quot;192.168.170.131&quot;);</div><div>        //端口 默认 5672</div><div>        factory.setPort(5672);</div><div>        //虚拟机,默认值 &quot;/&quot;</div><div>        factory.setVirtualHost(&quot;/myRabbit&quot;);</div><div>        //用户名密码</div><div>        factory.setUsername(&quot;gzx&quot;);</div><div>        factory.setPassword(&quot;gzx&quot;);</div><div>        //3. 创建连接Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div>        //5. 创建交换机</div><div>        /**</div><div>         * exchangeDeclare(String exchange, 交换机名称</div><div>         *         BuiltinExchangeType type, String类型,或者枚举类型 (direct 定向发送消息,fanout 广播发送,</div><div>         *         topic 通配符方式,headers 参数匹配)</div><div>         *         boolean durable, 是否持久化</div><div>         *         boolean autoDelete, 当没有消费者连接后是否自动删除</div><div>         *         boolean internal,内部的, MQ使用,比如erlang语言内部插件调用</div><div>         *         Map&lt;String, Object&gt; arguments) 参数</div><div>         */</div><div>        String exchangeName =&quot;HelloWorld_fanout&quot;;</div><div>        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,null);</div><div>        //6. 创建队列</div><div>        /**</div><div>         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,</div><div>         *                                  Map&lt;String, Object&gt; arguments);</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. durable:是否持久化</div><div>         *      3. exclusive:</div><div>         *              是否独占,只能有一个消费者监听这个队列</div><div>         *              当Connection关闭时,是否删除队列</div><div>         *      4. autoDelete:是否自动删除,当没有Consumer来消费时,自动删除掉,</div><div>         *      5. arguments:参数.</div><div>         */</div><div>        String queue1Name = &quot;PubSub_queue1&quot;;</div><div>        String queue2Name = &quot;PubSub_queue2&quot;;</div><div>        //如果没有一个名字叫hello_world的队列,就创建它,有则不创建</div><div>        channel.queueDeclare(queue1Name,true,false,false,null);</div><div>        channel.queueDeclare(queue2Name,true,false,false,null);</div><div>        //7. 队列和交换机绑定</div><div>        /**</div><div>         * queueBind(String queue, 队列名称</div><div>         *          String exchange, 交换机名称</div><div>         *          String routingKey 路由键 ,即绑定规则 ,fanout 模式下 routingKey为&quot;&quot;)</div><div>         */</div><div>        channel.queueBind(queue1Name,exchangeName,&quot;&quot;);</div><div>        channel.queueBind(queue2Name,exchangeName,&quot;&quot;);</div><div>        //6. 发送消息</div><div>        /**</div><div>         * basicPublish(String exchange, String routingKey, boolean mandatory,</div><div>         *                          boolean immediate, AMQP.BasicProperties props, byte[] body)</div><div>         * 参数:</div><div>         *      1. exchange:交换机名称,简单模式下,只能使用默认交换机</div><div>         *      2. routingKey:路由名称</div><div>         *      3. prop:配置信息</div><div>         *      4. body:发送的消息数据,</div><div>         */</div><div>        int msgNumber =10;</div><div>        for (int i = 1; i &lt;= msgNumber ; i++) {</div><div>            String body =&quot;PubSub _msg&quot;+i;</div><div>            //使用了默认的交换机,routingKey就要和队列的名称一样,才能路由到对应的队列中</div><div>            channel.basicPublish(exchangeName,&quot;&quot;,null,body.getBytes());</div><div>        }</div><div>        channel.close();</div><div>        connection.close();</div><div>    }</div><div>}</div></div><div>消费者1代码</div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div>/**</div><div>* 接受消息</div><div>* @author GaoZeXi</div><div>*/</div><div>public class PubSubConsumer1 {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div>        //1.创建连接工厂,获取连接</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2.设置参数 ,不设置则使用默认值localhost</div><div>        factory.setHost(&quot;192.168.170.131&quot;);</div><div>        //端口 默认 5672</div><div>        factory.setPort(5672);</div><div>        //虚拟机,默认值 &quot;/&quot;</div><div>        factory.setVirtualHost(&quot;/myRabbit&quot;);</div><div>        //用户名密码</div><div>        factory.setUsername(&quot;gzx&quot;);</div><div>        factory.setPassword(&quot;gzx&quot;);</div><div>        //3. 创建连接Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div>        //5. 创建队列</div><div>        /**</div><div>         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,</div><div>         *                                  Map&lt;String, Object&gt; arguments);</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. durable:是否持久化</div><div>         *      3. exclusive:</div><div>         *              是否独占,只能有一个消费者监听这个队列</div><div>         *              当Connection关闭时,是否删除队列</div><div>         *      4. autoDelete:是否自动删除,当没有Consumer来消费时,自动删除掉,</div><div>         *      5. arguments:参数.</div><div>         */</div><div>        //如果没有一个名字叫hello_world的队列,就创建它,有则不创建</div><div>        //channel.queueDeclare(&quot;work_queues&quot;,true,false,false,null);</div><div>        //6. 接受消息</div><div>        /**</div><div>         * basicConsume(String queue, boolean autoAck, Consumer callback)</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. autoAck:是否自动确认</div><div>         *      3. callback:回调对象</div><div>         */</div><div>        String queue1Name = &quot;PubSub_queue1&quot;;</div><div><br/></div><div><br/></div><div>        Consumer consumer = new DefaultConsumer(channel){</div><div>            /**</div><div>             * 回调方法,当收到消息后,会自动执行该方法</div><div>             * @param consumerTag 消息标识</div><div>             * @param envelope 获取一些信息,比如 交换机,routingKey...</div><div>             * @param properties 配置信息</div><div>             * @param body 消息体</div><div>             * @throws IOException</div><div>             */</div><div>            @Override</div><div>            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {</div><div>//                System.out.println(&quot;consumerTag:&quot;+consumerTag);</div><div>//                System.out.println(&quot;getExchange:&quot;+envelope.getExchange());</div><div>//                System.out.println(&quot;getRoutingKey:&quot;+envelope.getRoutingKey());</div><div>//                System.out.println(&quot;getDeliveryTag:&quot;+envelope.getDeliveryTag());</div><div>//                System.out.println(&quot;properties:&quot;+properties);</div><div>                System.out.println(&quot;body保存在数据库:&quot;+new String(body));</div><div>            }</div><div>        };</div><div>        //监听队列.消费数据</div><div>        channel.basicConsume(queue1Name,true,consumer);</div><div>//        消费者需要一直监听消息队列,不需要关闭!</div><div>//        channel.close();</div><div>//        connection.close();</div><div>    }</div><div>}</div></div><div>消费者2代码</div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div>/**</div><div>* 接受消息</div><div>* @author GaoZeXi</div><div>*/</div><div>public class PubSubConsumer2 {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div>        //1.创建连接工厂,获取连接</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2.设置参数 ,不设置则使用默认值localhost</div><div>        factory.setHost(&quot;192.168.170.131&quot;);</div><div>        //端口 默认 5672</div><div>        factory.setPort(5672);</div><div>        //虚拟机,默认值 &quot;/&quot;</div><div>        factory.setVirtualHost(&quot;/myRabbit&quot;);</div><div>        //用户名密码</div><div>        factory.setUsername(&quot;gzx&quot;);</div><div>        factory.setPassword(&quot;gzx&quot;);</div><div>        //3. 创建连接Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div>        //5. 创建队列</div><div>        /**</div><div>         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,</div><div>         *                                  Map&lt;String, Object&gt; arguments);</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. durable:是否持久化</div><div>         *      3. exclusive:</div><div>         *              是否独占,只能有一个消费者监听这个队列</div><div>         *              当Connection关闭时,是否删除队列</div><div>         *      4. autoDelete:是否自动删除,当没有Consumer来消费时,自动删除掉,</div><div>         *      5. arguments:参数.</div><div>         */</div><div>        //如果没有一个名字叫hello_world的队列,就创建它,有则不创建</div><div>        //channel.queueDeclare(&quot;work_queues&quot;,true,false,false,null);</div><div>        //6. 接受消息</div><div>        /**</div><div>         * basicConsume(String queue, boolean autoAck, Consumer callback)</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. autoAck:是否自动确认</div><div>         *      3. callback:回调对象</div><div>         */</div><div>        String queue2Name = &quot;PubSub_queue2&quot;;</div><div>        Consumer consumer = new DefaultConsumer(channel){</div><div>            /**</div><div>             * 回调方法,当收到消息后,会自动执行该方法</div><div>             * @param consumerTag 消息标识</div><div>             * @param envelope 获取一些信息,比如 交换机,routingKey...</div><div>             * @param properties 配置信息</div><div>             * @param body 消息体</div><div>             * @throws IOException</div><div>             */</div><div>            @Override</div><div>            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {</div><div>//                System.out.println(&quot;consumerTag:&quot;+consumerTag);</div><div>//                System.out.println(&quot;getExchange:&quot;+envelope.getExchange());</div><div>//                System.out.println(&quot;getRoutingKey:&quot;+envelope.getRoutingKey());</div><div>//                System.out.println(&quot;getDeliveryTag:&quot;+envelope.getDeliveryTag());</div><div>//                System.out.println(&quot;properties:&quot;+properties);</div><div>                System.out.println(&quot;body 打印到控制台:&quot;+new String(body));</div><div>            }</div><div>        };</div><div>        channel.basicConsume(queue2Name,true,consumer);</div><div>//        消费者需要一直监听消息队列,不需要关闭!</div><div>//        channel.close();</div><div>//        connection.close();</div><div>    }</div></div><div>效果</div><div><img src="Evernote_files/Image [10].png" type="image/png" data-filename="Image.png"/><img src="Evernote_files/Image [11].png" type="image/png" data-filename="Image.png"/></div><div><br/></div></span>
</div>
<hr>
<a name="781"/>

<div>
<span><div><img src="Evernote_files/Image [12].png" type="image/png" data-filename="Image.png"/></div><div><font style="font-size: 17pt;"><br/></font></div><div>demo 结构图</div><div><img src="Evernote_files/Image [13].png" type="image/png" data-filename="Image.png"/></div><div><font style="font-size: 17pt;">生产者代码:</font></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div>/**</div><div>* 发送消息</div><div>* @author GaoZeXi</div><div>*/</div><div>public class WorkQueuesProducer {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div>        //1.创建连接工厂,获取连接</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2.设置参数 ,不设置则使用默认值localhost</div><div>        factory.setHost(&quot;192.168.170.131&quot;);</div><div>        //端口 默认 5672</div><div>        factory.setPort(5672);</div><div>        //虚拟机,默认值 &quot;/&quot;</div><div>        factory.setVirtualHost(&quot;/myRabbit&quot;);</div><div>        //用户名密码</div><div>        factory.setUsername(&quot;gzx&quot;);</div><div>        factory.setPassword(&quot;gzx&quot;);</div><div>        //3. 创建连接Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div>        //5. 创建队列</div><div>        /**</div><div>         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,</div><div>         *                                  Map&lt;String, Object&gt; arguments);</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. durable:是否持久化</div><div>         *      3. exclusive:</div><div>         *              是否独占,只能有一个消费者监听这个队列</div><div>         *              当Connection关闭时,是否删除队列</div><div>         *      4. autoDelete:是否自动删除,当没有Consumer来消费时,自动删除掉,</div><div>         *      5. arguments:参数.</div><div>         */</div><div>        //如果没有一个名字叫hello_world的队列,就创建它,有则不创建</div><div>        channel.queueDeclare(&quot;work_queues&quot;,true,false,false,null);</div><div>        //6. 发送消息</div><div>        /**</div><div>         * basicPublish(String exchange, String routingKey, boolean mandatory,</div><div>         *                          boolean immediate, AMQP.BasicProperties props, byte[] body)</div><div>         * 参数:</div><div>         *      1. exchange:交换机名称,简单模式下,只能使用默认交换机</div><div>         *      2. routingKey:路由名称</div><div>         *      3. prop:配置信息</div><div>         *      4. body:发送的消息数据,</div><div>         */</div><div>        int msgNumber =10;</div><div>        for (int i = 1; i &lt;= msgNumber ; i++) {</div><div>            String body =&quot;HelloWorld WorkQueues _msg&quot;+i;</div><div>            //使用了默认的交换机,routingKey就要和队列的名称一样,才能路由到对应的队列中</div><div>            channel.basicPublish(&quot;&quot;,&quot;work_queues&quot;,null,body.getBytes());</div><div>        }</div><div>        channel.close();</div><div>        connection.close();</div><div>    }</div></div><div><span style="font-size: 13pt;"><br/></span></div><div><span style="font-size: 17pt;">消费者代码:</span></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div>/**</div><div>* 接受消息</div><div>* @author GaoZeXi</div><div>*/</div><div>public class WorkQueuesConsumer1 {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div>        //1.创建连接工厂,获取连接</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2.设置参数 ,不设置则使用默认值localhost</div><div>        factory.setHost(&quot;192.168.170.131&quot;);</div><div>        //端口 默认 5672</div><div>        factory.setPort(5672);</div><div>        //虚拟机,默认值 &quot;/&quot;</div><div>        factory.setVirtualHost(&quot;/myRabbit&quot;);</div><div>        //用户名密码</div><div>        factory.setUsername(&quot;gzx&quot;);</div><div>        factory.setPassword(&quot;gzx&quot;);</div><div>        //3. 创建连接Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div>        //5. 创建队列</div><div>        /**</div><div>         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,</div><div>         *                                  Map&lt;String, Object&gt; arguments);</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. durable:是否持久化</div><div>         *      3. exclusive:</div><div>         *              是否独占,只能有一个消费者监听这个队列</div><div>         *              当Connection关闭时,是否删除队列</div><div>         *      4. autoDelete:是否自动删除,当没有Consumer来消费时,自动删除掉,</div><div>         *      5. arguments:参数.</div><div>         */</div><div>        //如果没有一个名字叫hello_world的队列,就创建它,有则不创建</div><div>        channel.queueDeclare(&quot;work_queues&quot;,true,false,false,null);</div><div>        //6. 接受消息</div><div>        /**</div><div>         * basicConsume(String queue, boolean autoAck, Consumer callback)</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. autoAck:是否自动确认</div><div>         *      3. callback:回调对象</div><div>         */</div><div>        Consumer consumer = new DefaultConsumer(channel){</div><div>            /**</div><div>             * 回调方法,当收到消息后,会自动执行该方法</div><div>             * @param consumerTag 消息标识</div><div>             * @param envelope 获取一些信息,比如 交换机,routingKey...</div><div>             * @param properties 配置信息</div><div>             * @param body 消息体</div><div>             * @throws IOException</div><div>             */</div><div>            @Override</div><div>            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {</div><div>//                System.out.println(&quot;consumerTag:&quot;+consumerTag);</div><div>//                System.out.println(&quot;getExchange:&quot;+envelope.getExchange());</div><div>//                System.out.println(&quot;getRoutingKey:&quot;+envelope.getRoutingKey());</div><div>//                System.out.println(&quot;getDeliveryTag:&quot;+envelope.getDeliveryTag());</div><div>//                System.out.println(&quot;properties:&quot;+properties);</div><div>                System.out.println(&quot;body:&quot;+new String(body));</div><div>            }</div><div>        };</div><div>        channel.basicConsume(&quot;work_queues&quot;,true,consumer);</div><div>//        消费者需要一直监听消息队列,不需要关闭!</div><div>//        channel.close();</div><div>//        connection.close();</div><div>    }</div></div><div><span style="font-size: 17.3333px;">消费者1和消费者2代码相同,互为竞争,同一个消息只能被消费一次</span></div><div><span style="font-size: 13pt;"><br/></span></div><div><span style="font-size: 13pt;">总结:</span></div><div><span style="font-size: 13pt;">      在一个队列中如果有多个消费者,那么消费者之间对于同一个消息的关系是竞争的关系.</span></div><div><span style="font-size: 13pt;">      Work Queues 对于人物过重或者任务较多的情况下使用工作队列可以提高任务处理的速度,例如:短信服务器部署多个,只需要有一个节点成功发送即可.</span></div></span>
</div>
<hr>
<a name="765"/>

<div>
<span><div><img src="Evernote_files/Image [14].png" type="image/png" data-filename="Image.png"/><img src="Evernote_files/Image [15].png" type="image/png" data-filename="Image.png"/></div><div><span style="font-size: 14pt;">1 创建工程 ,生产者和消费者的pom文件相同</span></div><div><br/></div><div><br/></div><div><img src="Evernote_files/Image [16].png" type="image/png" data-filename="Image.png"/></div><div><br/></div><div><span style="font-size: 14pt;">2 编写生产者代码</span></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div>/**</div><div>* 发送消息</div><div>* @author GaoZeXi</div><div>*/</div><div>public class HelloWorld {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div>        //1.创建连接工厂,获取连接</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2.设置参数 ,不设置则使用默认值localhost</div><div>        factory.setHost(&quot;192.168.170.131&quot;);</div><div>        //端口 默认 5672</div><div>        factory.setPort(5672);</div><div>        //虚拟机,默认值 &quot;/&quot;</div><div>        factory.setVirtualHost(&quot;/myRabbit&quot;);</div><div>        //用户名密码</div><div>        factory.setUsername(&quot;gzx&quot;);</div><div>        factory.setPassword(&quot;gzx&quot;);</div><div>        //3. 创建连接Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div>        //5. 创建队列</div><div>        /**</div><div>         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,</div><div>         *                                  Map&lt;String, Object&gt; arguments);</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. durable:是否持久化</div><div>         *      3. exclusive:</div><div>         *              是否独占,只能有一个消费者监听这个队列</div><div>         *              当Connection关闭时,是否删除队列</div><div>         *      4. autoDelete:是否自动删除,当没有Consumer来消费时,自动删除掉,</div><div>         *      5. arguments:参数.</div><div>         */</div><div>        //如果没有一个名字叫hello_world的队列,就创建它,有则不创建</div><div>        channel.queueDeclare(&quot;hello_world&quot;,true,false,false,null);</div><div>        //6. 发送消息</div><div>        /**</div><div>         * basicPublish(String exchange, String routingKey, boolean mandatory,</div><div>         *                          boolean immediate, AMQP.BasicProperties props, byte[] body)</div><div>         * 参数:</div><div>         *      1. exchange:交换机名称,简单模式下,只能使用默认交换机</div><div>         *      2. routingKey:路由名称</div><div>         *      3. prop:配置信息</div><div>         *      4. body:发送的消息数据,</div><div>         */</div><div>        String body =&quot;helloWorld_msg&quot;;</div><div>        //使用了默认的交换机,routingKey就要和队列的名称一样,才能路由到对应的队列中</div><div>        channel.basicPublish(&quot;&quot;,&quot;hello_world&quot;,null,body.getBytes());</div><div>        channel.close();</div><div>        connection.close();</div><div>    }</div><div>}</div></div><div><span style="font-size: 15pt;">3 编写消费者代码</span></div><div style="box-sizing: border-box; padding: 8px; font-family: Monaco, Menlo, Consolas, &quot;Courier New&quot;, monospace; font-size: 12px; color: rgb(51, 51, 51); border-radius: 4px; background-color: rgb(251, 250, 248); border: 1px solid rgba(0, 0, 0, 0.15);-en-codeblock:true;"><div><br/></div><div><br/></div><div>/**</div><div>* 接受消息</div><div>* @author GaoZeXi</div><div>*/</div><div>public class HelloWorldConsumer {</div><div>    public static void main(String[] args) throws IOException, TimeoutException {</div><div>        //1.创建连接工厂,获取连接</div><div>        ConnectionFactory factory = new ConnectionFactory();</div><div>        //2.设置参数 ,不设置则使用默认值localhost</div><div>        factory.setHost(&quot;192.168.170.131&quot;);</div><div>        //端口 默认 5672</div><div>        factory.setPort(5672);</div><div>        //虚拟机,默认值 &quot;/&quot;</div><div>        factory.setVirtualHost(&quot;/myRabbit&quot;);</div><div>        //用户名密码</div><div>        factory.setUsername(&quot;gzx&quot;);</div><div>        factory.setPassword(&quot;gzx&quot;);</div><div>        //3. 创建连接Connection</div><div>        Connection connection = factory.newConnection();</div><div>        //4. 创建Channel</div><div>        Channel channel = connection.createChannel();</div><div>        //5. 创建队列</div><div>        /**</div><div>         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,</div><div>         *                                  Map&lt;String, Object&gt; arguments);</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. durable:是否持久化</div><div>         *      3. exclusive:</div><div>         *              是否独占,只能有一个消费者监听这个队列</div><div>         *              当Connection关闭时,是否删除队列</div><div>         *      4. autoDelete:是否自动删除,当没有Consumer来消费时,自动删除掉,</div><div>         *      5. arguments:参数.</div><div>         */</div><div>        //如果没有一个名字叫hello_world的队列,就创建它,有则不创建</div><div>        channel.queueDeclare(&quot;hello_world&quot;,true,false,false,null);</div><div>        //6. 接受消息</div><div>        /**</div><div>         * basicConsume(String queue, boolean autoAck, Consumer callback)</div><div>         * 参数:</div><div>         *      1. queue:队列名</div><div>         *      2. autoAck:是否自动确认</div><div>         *      3. callback:回调对象</div><div>         */</div><div>        Consumer consumer = new DefaultConsumer(channel){</div><div>            /**</div><div>             * 回调方法,当收到消息后,会自动执行该方法</div><div>             * @param consumerTag 消息标识</div><div>             * @param envelope 获取一些信息,比如 交换机,routingKey...</div><div>             * @param properties 配置信息</div><div>             * @param body 消息体</div><div>             * @throws IOException</div><div>             */</div><div>            @Override</div><div>            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {</div><div>                System.out.println(&quot;consumerTag:&quot;+consumerTag);</div><div>                System.out.println(&quot;getExchange:&quot;+envelope.getExchange());</div><div>                System.out.println(&quot;getRoutingKey:&quot;+envelope.getRoutingKey());</div><div>                System.out.println(&quot;getDeliveryTag:&quot;+envelope.getDeliveryTag());</div><div>                System.out.println(&quot;properties:&quot;+properties);</div><div>                System.out.println(&quot;body:&quot;+new String(body));</div><div>            }</div><div>        };</div><div>        channel.basicConsume(&quot;hello_world&quot;,true,consumer);</div><div>//        消费者需要一直监听消息队列,不需要关闭!</div><div>//        channel.close();</div><div>//        connection.close();</div><div>    }</div><div>}</div></div><div><span style="font-size: 13pt;"><br/></span></div></span>
</div>
<hr>
<a name="771"/>

<div>
<span><div>rabbitMQ 消息发送流程图</div><div><img src="Evernote_files/Image [17].png" type="image/png" data-filename="Image.png"/></div><div><br/></div></span>
</div>
<hr>
<a name="747"/>

<div>
<span><div><img src="Evernote_files/Image [18].png" type="image/png" data-filename="Image.png"/></div><div><font style="font-size: 16pt;">ADMIN界面</font></div><div><img src="Evernote_files/Image [19].png" type="image/png" data-filename="Image.png"/></div><div><font style="font-size: 16pt;">Overview</font></div><div><img src="Evernote_files/Image [20].png" type="image/png" data-filename="Image.png"/></div><div><img src="Evernote_files/Image [21].png" type="image/png" data-filename="Image.png"/></div><div><br/></div></span>
</div></body></html> 